package org.jkuang.qstar.index.core;

import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.log4j.Logger;
import org.jkuang.qstar.commons.io.FileReader;
import org.jkuang.qstar.commons.io.FileWriter;
import org.jkuang.qstar.commons.jni.CurMap;
import org.jkuang.qstar.index.cmp.IndexObject;
import org.jkuang.qstar.index.cmp.KeyComparator;
import org.jkuang.qstar.index.cmp.SortComparator;
import org.jkuang.qstar.index.cmp.ValueComparator;
import org.jkuang.qstar.index.dic.TermsDic;
import org.jkuang.qstar.index.iterator.TSIterator;
import org.jkuang.qstar.index.store.DataNode;
import org.jkuang.qstar.index.store.DataStruct;
/**
 * 
 * @author jkuang 2012年11月17日
 *
 */
public class IndexWriter {
	public static String ALL = "_all:*";
	// 标记文件结束
	public static byte FILE_EOF = 0x0E;
	// 标记数据插入
	public static byte NODE_ADD = 0x0A;
	// 文件起始标记
	public static long HEAD_MARK = 0x0123454321L;
	// 文件终止标记
	public static long TAIL_MARK = 0x9876543210L;

	public static Charset charset = Charset.forName("utf8");

	private TermsDic dic;
	private CurMap store;
	private DataStruct<DataNode> keys;
	private DataStruct<IndexObject> indexs;
	private SortComparator sortComparator = new SortComparator();
	private Logger logger = Logger.getLogger("IndexWriter");
	private ReentrantReadWriteLock rwLock;
	// 索引下面的锁
	private ReentrantReadWriteLock indexLock = new ReentrantReadWriteLock();
    private ValueComparator vcomparator = new ValueComparator();
	public IndexWriter(ReentrantReadWriteLock rwLock, TermsDic dic, DataStruct<IndexObject> indexs, CurMap store) {

		this.dic = dic;
		this.indexs = indexs;
		this.store = store;
		this.rwLock = rwLock;
		this.keys = new DataStruct<DataNode>(rwLock, new KeyComparator(), 0);
		this.indexs.insert(new DataStruct<DataNode>(indexLock, sortComparator, this.dic.add(ALL)));
	}

	/**
	 * 插入数据节点
	 * 
	 * @param key   数据主键
	 * @param sort  排序字段，通常采用时间戳
	 * @param terms 需要被索引的词条
	 * @param data  需要返回的数据
	 */
	public void insert(long key, long sort, Collection<String> terms, byte[] data) {
		rwLock.writeLock().lock();
		try {
			insert(addToDic(key, sort, terms), data);
		} finally {
			rwLock.writeLock().unlock();
		}
	}

	/**
	 * 删除数据节点
	 * 
	 * @param key 需要删除的数据主键
	 */
	public void delete(long key) {
		rwLock.writeLock().lock();
		try {
			this.delete(DataNode.create(key, 0));
			this.store.remove(key);
		} finally {
			rwLock.writeLock().unlock();
		}
	}


	private boolean insert(DataNode node, byte[] newData) {
		DataNode temp = keys.get(node);
		// 新旧数据完全一样
		byte[] oldData = this.store.get(node.key());

		// 新旧索引完全一致
		if (temp != null && temp.sort() == node.sort() && vcomparator.compare(temp.propertys(), node.propertys())) {

			// 新旧都没有存储数据
			if ( vcomparator.compare(oldData, newData))// 新旧存储数据完全一样
			{
				return false;
			}
			if (! vcomparator.isEmpty(newData)) {//新数据与旧数据不相同
				this.store.put(node.key(), newData);
			} else if (! vcomparator.isEmpty(oldData)) {//数据被删除了
				this.store.remove(node.key());
			}
			return true;
		}

		if (temp != null) {
			this.delete(temp);
		}
		if (! vcomparator.isEmpty(newData) && ! vcomparator.compare(oldData, newData)) {
			this.store.put(node.key(), newData);
		} else if (! vcomparator.isEmpty(oldData)) {
			this.store.remove(node.key());
		}
		this.keys.insert(node);
		IndexObject object = new IndexObject();
		for (int property : node.propertys()) {
			@SuppressWarnings("unchecked")
			DataStruct<DataNode> index = (DataStruct<DataNode>) indexs.get(object.setIndex(property));
			if (index == null) {
				index = new DataStruct<DataNode>(indexLock, sortComparator, property);
				this.indexs.insert(index);
			}
			index.insert(node);
		}
		return true;
	}

	@SuppressWarnings("unchecked")
	private void delete(final DataNode temp) {
		DataNode node = keys.get(temp);
		if (node != null) {
			for (int property : node.propertys()) {
				DataStruct<DataNode> index = (DataStruct<DataNode>) indexs.get(new IndexObject().setIndex(property));
				if (index != null) {
					index.remove(node);
					if (index.scanSize(null, null) == 0) {
						this.indexs.remove(index);
					}
				}
			}
			keys.remove(node);
		}
	}

	/**
	 * 将目标字段值添加到词典内
	 * 
	 * @param key
	 * @param sort
	 * @param terms
	 * @return
	 */
	DataNode addToDic(long key, long sort, Collection<String> terms) {
		if (terms == null) {
			return DataNode.create(key, sort, null);
		} else {
			terms.add(ALL);
			Set<Integer> values = new TreeSet<Integer>();
			for (String term : terms) {
				values.add(this.dic.add(term));
			}
			int[] array = new int[values.size()];
			int newIndex = 0;
			for (int i : values) {
				array[newIndex++] = i;
			}
			return DataNode.create(key, sort, array);
		}
	}

	/**
	 * 对索引数组进行冒泡排序
	 * 
	 * @param arr
	 * @return
	 */
	int[] bubbleSort(int[] arr) {
		for (int i = 0; i < arr.length - 1; i++) {
			boolean flag = true;
			for (int j = 0; j < arr.length - 1 - i; j++) {
				if (arr[j] > arr[j + 1]) {
					int v = arr[j];
					arr[j] = arr[j + 1];
					arr[j + 1] = v;
					flag = false;
				}
			}
			if (flag) {
				break;
			}
		}
		return arr;
	}

	/**
	 * dump 文件 dump ,delete,add 是互斥的，单与读可共存，此处需要加读锁
	 * 
	 * @param dic 持久化的文件夹，若不存在，则会新创建
	 * 
	 * @throws IOException
	 */
	@SuppressWarnings("unchecked")
	public void dumpSnapshot(File dir) throws IOException {
		rwLock.readLock().lock();
		try {
			DataStruct<DataNode> index = (DataStruct<DataNode>) indexs
					.get(new IndexObject().setIndex(this.dic.get(ALL)));
			TSIterator<DataNode> iterator = index.blockList(null, null, false, null);
			if (iterator == null) {
				logger.warn("内存中没有任何数据！");
				return;
			}
			if (!dir.exists()) {
				dir.mkdirs();
			}
			File tempfile = new File(dir, "snapshot.bin.temp");
			FileWriter writer = new FileWriter(tempfile, false);
			try {
				writer.writeLong(HEAD_MARK);
				writer.writeLong(System.currentTimeMillis());
				int dicSize = this.dic.size();
				writer.writeInt(dicSize);
				for (int i = 0; i < dicSize; i++) {
					writer.writeShortBytes(this.dic.get(i).getBytes(charset));
				}
				while (iterator.hasNext()) {
					DataNode node = iterator.next();
					writer.writeByte(NODE_ADD);
					writer.writeLong(node.key());
					writer.writeLong(node.sort());
					int[] array = node.propertys();
					writer.writeShort((short) array.length);
					for (int i = 0; i < (short) array.length; i++) {
						writer.writeInt(array[i]);
					}
					byte[] data = this.store.get(node.key());
					writer.writeIntBytes(data);
				}
				writer.writeByte(FILE_EOF);
				writer.writeLong(TAIL_MARK);
				writer.flush();
				writer.close();
				File dest = new File(dir, "snapshot.bin");
				dest.delete();
				tempfile.renameTo(dest);
			} catch (Exception e) {
				e.printStackTrace();
			}
		} finally {
			rwLock.readLock().unlock();
		}

	}

	/**
	 * 校验dump文件是否有损坏，是否已过期
	 * 
	 * @param dir 存放文件袋额目录
	 * @return 校验成功则返回true
	 * @throws IOException
	 */
	@SuppressWarnings("unused")
	public boolean checkSnapshot(File dir) throws IOException {
		File file = new File(dir, "snapshot.bin");
		if (!file.exists()) {
			logger.info("file not exists! path:" + file.getAbsolutePath());
			return false;
		}
		FileReader reader = new FileReader(file);
		try {
			if (reader.hasMore(20) && reader.readLong() == HEAD_MARK) {
				long time = reader.readLong();
				if (System.currentTimeMillis() - time > 86400l * 1000l * 2) {
					logger.error("文件生成时间超过两天,已过期！");
					return false;
				}
				int size = reader.readInt();
				for (int i = 0; i < size; i++) {
					if (reader.hasMore(2)) {
						int length = reader.readShort();
						if (length == 0) {
							logger.error("词条校验失败，词条长度为0");
							return false;
						}
						if (reader.hasMore(length)) {
							reader.readBytes(length);
						} else {
							logger.error("词条校验失败，文件已到结尾");
							return false;
						}
					} else {
						logger.error("词条校验失败，文件已到结尾");
						return false;
					}
				}
				int count = 0;
				while (true) {
					if (reader.hasMore(9)) {
						byte mark = reader.readByte();
						long key = reader.readLong();
						if (mark == NODE_ADD && reader.hasMore(10)) {
							count++;
							long sort = reader.readLong();
							int arraySize = reader.readShort();
							if (reader.hasMore(arraySize * 4 + 4)) {
								for (int i = 0; i < arraySize; i++) {
									reader.readInt();
								}
								int length = reader.readInt();
								if (reader.hasMore(length)) {
									reader.readBytes(length);
								} else {
									// 数据存储错误
									logger.error("文件在校验第 " + count + "条时失败，该文档数据读取错误！key:" + key);
									return false;
								}
							} else {
								logger.error("文件在校验第 " + count + "条时失败，该文档词条索引读取错误！key:" + key);
								return false;
							}
						} else if (mark == FILE_EOF && key == TAIL_MARK) {
							// localOK
							logger.info("校验成功，文件完整，可加装！");
							System.out.println("total count:" + count);
							return true;
						} else {
							logger.error("文件在校验第 " + (count + 1) + "条时失败，文件dump异常，数据不完整！key:" + key);
							return false;
						}
					} else {
						return false;
					}
				}

			} else {
				logger.error("文件校验失败，文件头校验不符");
				return false;
			}
		} finally {
			reader.close();
		}

	}

	/**
	 * 加载文件
	 * 
	 * @param dir 存文dump文件的文件夹，dump命名为 snapshot.bin
	 * @return dump成功返回true，否则返回false
	 * @throws IOException
	 */
	public boolean loadSnapshot(File dir) throws IOException {
		File file = new File(dir, "snapshot.bin");
		if (!file.exists()) {
			return false;
		}
		FileReader reader = new FileReader(file);
		try {
			if (reader.hasMore(20) && reader.readLong() == HEAD_MARK) {
				long time = reader.readLong();
				if (System.currentTimeMillis() - time > 86400 * 1000l * 2) {
					return false;
				}
				int size = reader.readInt();
				Map<Integer, Integer> map = new HashMap<Integer, Integer>();
				for (int i = 0; i < size; i++) {
					if (reader.hasMore(2)) {
						int length = reader.readShort();
						if (reader.hasMore(length)) {
							byte[] temp = reader.readBytes(length);
							// 将旧的index映射到新词典中的index
							map.put(i, this.dic.add(new String(temp, charset)));
						} else {
							return false;
						}
					} else {
						return false;
					}
				}
				int count = 0;
				while (true) {

					if (reader.hasMore(9)) {
						byte mark = reader.readByte();
						long key = reader.readLong();
						if (mark == NODE_ADD && reader.hasMore(10)) {

							long sort = reader.readLong();
							int arraySize = reader.readShort();
							if (reader.hasMore(arraySize * 4 + 4)) {
								int[] field = new int[arraySize];
								for (int i = 0; i < field.length; i++) {
									field[i] = map.get(reader.readInt());
								}

								if (count++ % 10000 == 10000 - 1) {
									System.out.println("aadd:" + count);
								}
								int dataLength = reader.readInt();
								if (reader.hasMore(dataLength)) {
									byte[] data = reader.readBytes(dataLength);
									// if(count>13000000)
									this.insert(DataNode.create(key, sort, bubbleSort(field)), data);
								} else {
									System.err.println("数据存储错误!");
									return false;
								}

							} else {
								return false;
							}
						} else if (mark == FILE_EOF && key == TAIL_MARK) {
							// localOK
							return true;
						} else {
							return false;
						}
					} else {
						return false;
					}
				}

			} else {
				return false;
			}
		} finally {
			reader.close();
		}
	}
}
